/*
 * Copyright 2011-2012 Amazon Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *    http://aws.amazon.com/apache2.0
 *
 * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES
 * OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and
 * limitations under the License.
 */
package com.swengle.phoebe.reflect;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

import com.amazonaws.services.dynamodb.datamodeling.DynamoDBAttribute;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBAutoGeneratedKey;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBHashKey;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBIgnore;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBMappingException;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBMarshaller;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBMarshalling;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBRangeKey;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBTable;
import com.amazonaws.services.dynamodb.datamodeling.DynamoDBVersionAttribute;
import com.amazonaws.services.dynamodb.model.AttributeValue;
import com.amazonaws.util.DateUtils;
import com.swengle.phoebe.annotation.DynamoDBTableInitialCapacities;
import com.swengle.phoebe.annotation.OnDelete;
import com.swengle.phoebe.annotation.OnRead;
import com.swengle.phoebe.annotation.OnCreate;
import com.swengle.phoebe.annotation.OnUpdate;

/**
 * Reflection assistant for {@link com.amazonaws.services.dynamodb.datamodeling.DynamoDBMapper}
 */
public class DynamoDBReflector {
	public static final DynamoDBReflector INSTANCE = new DynamoDBReflector();

	private DynamoDBReflector() {
	}

	/*
	 * Several caches for performance. Collectively, they can make this class
	 * over twice as fast.
	 */
	private final Map<Class<?>, Collection<Method>> getterCache = new HashMap<Class<?>, Collection<Method>>();
	private final Map<Class<?>, Method> hashKeyGetterCache = new HashMap<Class<?>, Method>();
	private final Map<Class<?>, Method> rangeKeyGetterCache = new HashMap<Class<?>, Method>();

	/** Methods in the hierarchy annotated with @OnCreate */
	private final Map<Class<?>, Collection<Method>> onCreateMethodCache = new HashMap<Class<?>, Collection<Method>>();

	/** Methods in the hierarchy annotated with @OnRead */
	private final Map<Class<?>, Collection<Method>> onReadMethodCache = new HashMap<Class<?>, Collection<Method>>();

	/** Methods in the hierarchy annotated with @OnUpdate */
	private final Map<Class<?>, Collection<Method>> onUpdateMethodCache = new HashMap<Class<?>, Collection<Method>>();

	/** Methods in the hierarchy annotated with @OnDelete */
	private final Map<Class<?>, Collection<Method>> onDeleteMethodCache = new HashMap<Class<?>, Collection<Method>>();

	/*
	 * All caches keyed by a Method use the getter for a particular mapped
	 * property
	 */
	private final Map<Method, Method> setterCache = new HashMap<Method, Method>();
	private final Map<Method, String> attributeNameCache = new HashMap<Method, String>();
	private final Map<Method, ArgumentUnmarshaller> argumentUnmarshallerCache = new HashMap<Method, ArgumentUnmarshaller>();
	private final Map<Method, ArgumentMarshaller> argumentMarshallerCache = new HashMap<Method, ArgumentMarshaller>();
	private final Map<Method, ArgumentMarshaller> versionArgumentMarshallerCache = new HashMap<Method, ArgumentMarshaller>();
	private final Map<Method, ArgumentMarshaller> keyArgumentMarshallerCache = new HashMap<Method, ArgumentMarshaller>();
	private final Map<Method, Boolean> versionAttributeGetterCache = new HashMap<Method, Boolean>();
	private final Map<Method, Boolean> autoGeneratedKeyGetterCache = new HashMap<Method, Boolean>();

	private final Map<Class<?>, Map<String, String>> fieldToAttributeNameCache = new HashMap<Class<?>, Map<String, String>>();
	private final Map<Method, MarshallerType> argumentMarshallerTypeCache = new HashMap<Method, MarshallerType>();

	/**
	 * Returns the set of getter methods which are relevant when marshalling or
	 * unmarshalling an object.
	 */
	public Collection<Method> getRelevantGetters(Class<?> clazz) {
		synchronized (getterCache) {
			if (!getterCache.containsKey(clazz)) {
				List<Method> relevantGetters = new LinkedList<Method>();
				for (Method m : clazz.getMethods()) {
					if (isRelevantGetter(m)) {
						relevantGetters.add(m);
					}
				}
				getterCache.put(clazz, relevantGetters);
			}
		}
		return getterCache.get(clazz);
	}

	/**
	 * Returns whether the method given is a getter method we should serialize /
	 * deserialize to the service. The method must begin with "get" or "is",
	 * have no arguments, belong to a class that declares its table, and not be
	 * marked ignored.
	 */
	private boolean isRelevantGetter(Method m) {
		return (m.getName().startsWith("get") || m.getName().startsWith("is"))
				&& m.getParameterTypes().length == 0
				&& m.getDeclaringClass().getAnnotation(DynamoDBTable.class) != null
				&& m.getAnnotation(DynamoDBIgnore.class) == null;
	}

	/**
	 * Returns the annotated {@link DynamoDBRangeKey} getter for the class
	 * given, or null if the class doesn't have one.
	 */
	public <T> Method getRangeKeyGetter(Class<T> clazz) {
		synchronized (rangeKeyGetterCache) {
			if (!rangeKeyGetterCache.containsKey(clazz)) {
				Method rangeKeyMethod = null;
				for (Method method : getRelevantGetters(clazz)) {
					if (method.getParameterTypes().length == 0
							&& method.getAnnotation(DynamoDBRangeKey.class) != null) {
						rangeKeyMethod = method;
						break;
					}
				}
				rangeKeyGetterCache.put(clazz, rangeKeyMethod);
			}
		}
		return rangeKeyGetterCache.get(clazz);
	}

	/**
	 * Returns the annotated {@link DynamoDBHashKey} getter for the class given,
	 * throwing an exception if there isn't one.
	 */
	public <T> Method getHashKeyGetter(Class<T> clazz) {
		synchronized (hashKeyGetterCache) {
			if (!hashKeyGetterCache.containsKey(clazz)) {
				for (Method method : getRelevantGetters(clazz)) {
					if (method.getParameterTypes().length == 0
							&& method.getAnnotation(DynamoDBHashKey.class) != null) {
						hashKeyGetterCache.put(clazz, method);
						break;
					}
				}
			}
		}

		Method hashKeyMethod = hashKeyGetterCache.get(clazz);
		if (hashKeyMethod == null) {
			throw new DynamoDBMappingException(
					"Public, zero-parameter hash key property must be annotated with "
							+ DynamoDBHashKey.class);
		}
		return hashKeyMethod;
	}

	/**
	 * Returns the {@link DynamoDBTable} annotation of the class given, throwing
	 * a runtime exception if it isn't annotated.
	 */
	public <T> DynamoDBTable getTable(Class<T> clazz) {
		DynamoDBTable table = clazz.getAnnotation(DynamoDBTable.class);
		if (table == null)
			throw new DynamoDBMappingException("Class " + clazz
					+ " must be annotated with " + DynamoDBTable.class);
		return table;
	}

	/**
	 * Returns whether or not this getter has a custom marshaller
	 */
	private boolean isCustomMarshaller(Method getter) {
		return getter.getAnnotation(DynamoDBMarshalling.class) != null;
	}

	/**
	 * Returns the argument unmarshaller used to unmarshall the getter / setter
	 * pair given.
	 * <p>
	 * Determining how to unmarshall a response, especially a numeric one,
	 * requires checking it against all supported types. This is expensive, so
	 * we cache a lookup table of getter method to argument unmarhsaller which
	 * can be reused.
	 * 
	 * @param toReturn
	 *            The typed domain object being unmarshalled for the client
	 * @param getter
	 *            The getter method being considered
	 * @param setter
	 *            The corresponding setter method being considered
	 */
	public <T> ArgumentUnmarshaller getArgumentUnmarshaller(final T toReturn,
			final Method getter, final Method setter) {
		synchronized (argumentUnmarshallerCache) {
			if (!argumentUnmarshallerCache.containsKey(getter)) {

				Class<?>[] parameterTypes = setter.getParameterTypes();
				Class<?> paramType = parameterTypes[0];
				if (parameterTypes.length != 1) {
					throw new DynamoDBMappingException(
							"Expected exactly one agument to " + setter);
				}

				ArgumentUnmarshaller unmarshaller = null;
				if (isCustomMarshaller(getter)) {
					unmarshaller = new SUnmarshaller() {

						@Override
						public Object unmarshall(AttributeValue value) {
							return getCustomMarshalledValue(toReturn, getter,
									value);
						}
					};
				} else {

					// If we're dealing with a collection, we need to get the
					// underlying type out of it
					boolean isCollection = false;
					if (Set.class.isAssignableFrom(paramType)) {
						isCollection = true;
						Type genericType = setter.getGenericParameterTypes()[0];
						if (genericType instanceof ParameterizedType) {
							paramType = (Class<?>) ((ParameterizedType) genericType)
									.getActualTypeArguments()[0];
						}
					} else if (Collection.class.isAssignableFrom(paramType)) {
						throw new DynamoDBMappingException(
								"Only java.util.Set collection types are permitted for "
										+ DynamoDBAttribute.class);
					}

					if (double.class.isAssignableFrom(paramType)
							|| Double.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<Double> argument = new HashSet<Double>();
									for (String s : value.getNS()) {
										argument.add(Double.parseDouble(s));
									}
									return argument;
								}

							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return Double.parseDouble(value.getN());
								}
							};
						}
					} else if (BigDecimal.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<BigDecimal> argument = new HashSet<BigDecimal>();
									for (String s : value.getNS()) {
										argument.add(new BigDecimal(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return new BigDecimal(value.getN());
								}
							};

						}
					} else if (BigInteger.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<BigInteger> argument = new HashSet<BigInteger>();
									for (String s : value.getNS()) {
										((Set<BigInteger>) argument)
												.add(new BigInteger(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return new BigInteger(value.getN());
								}
							};
						}
					} else if (int.class.isAssignableFrom(paramType)
							|| Integer.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<Integer> argument = new HashSet<Integer>();
									for (String s : value.getNS()) {
										argument.add(Integer.parseInt(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return Integer.parseInt(value.getN());
								}
							};
						}
					} else if (float.class.isAssignableFrom(paramType)
							|| Float.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<Float> argument = new HashSet<Float>();
									for (String s : value.getNS()) {
										argument.add(Float.parseFloat(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return Float.parseFloat(value.getN());
								}
							};
						}
					} else if (byte.class.isAssignableFrom(paramType)
							|| Byte.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<Byte> argument = new HashSet<Byte>();
									for (String s : value.getNS()) {
										argument.add(Byte.parseByte(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return Byte.parseByte(value.getN());
								}
							};
						}
					} else if (long.class.isAssignableFrom(paramType)
							|| Long.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<Long> argument = new HashSet<Long>();
									for (String s : value.getNS()) {
										argument.add(Long.parseLong(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return Long.parseLong(value.getN());
								}
							};
						}
					} else if (boolean.class.isAssignableFrom(paramType)
							|| Boolean.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new NSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<Boolean> argument = new HashSet<Boolean>();
									for (String s : value.getNS()) {
										argument.add(parseBoolean(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new NUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return parseBoolean(value.getN());
								}
							};
						}
					} else if (Date.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new SSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value)
										throws ParseException {
									Set<Date> argument = new HashSet<Date>();
									for (String s : value.getSS()) {
										argument.add(new DateUtils()
												.parseIso8601Date(s));
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new SUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value)
										throws ParseException {
									return new DateUtils()
											.parseIso8601Date(value.getS());
								}
							};
						}
					} else if (Calendar.class.isAssignableFrom(paramType)) {
						if (isCollection) {
							unmarshaller = new SSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value)
										throws ParseException {
									Set<Calendar> argument = new HashSet<Calendar>();
									for (String s : value.getSS()) {
										Calendar cal = GregorianCalendar
												.getInstance();
										cal.setTime(new DateUtils()
												.parseIso8601Date(s));
										argument.add(cal);
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new SUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value)
										throws ParseException {
									Calendar cal = GregorianCalendar
											.getInstance();
									cal.setTime(new DateUtils()
											.parseIso8601Date(value.getS()));
									return cal;
								}
							};
						}
					}

					/*
					 * After checking all other supported types, enforce a
					 * String match
					 */
					else if (!String.class.isAssignableFrom(paramType)) {
						throw new DynamoDBMappingException(
								"Expected a String, but was " + paramType);
					} else {
						if (isCollection) {
							unmarshaller = new SSUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									Set<String> argument = new HashSet<String>();
									for (String s : value.getSS()) {
										argument.add(s);
									}
									return argument;
								}
							};
						} else {
							unmarshaller = new SUnmarshaller() {

								@Override
								public Object unmarshall(AttributeValue value) {
									return value.getS();
								}
							};
						}
					}
				}

				argumentUnmarshallerCache.put(getter, unmarshaller);
			}
		}

		return argumentUnmarshallerCache.get(getter);
	}

	/**
	 * Marshalls the custom value given into the proper return type.
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	private <T> T getCustomMarshalledValue(T toReturn, Method getter,
			AttributeValue value) {
		DynamoDBMarshalling annotation = getter
				.getAnnotation(DynamoDBMarshalling.class);
		Class<? extends DynamoDBMarshaller<? extends Object>> marshallerClass = annotation
				.marshallerClass();

		DynamoDBMarshaller marshaller;
		try {
			marshaller = marshallerClass.newInstance();
		} catch (InstantiationException e) {
			throw new DynamoDBMappingException(
					"Couldn't instantiate marshaller of class "
							+ marshallerClass, e);
		} catch (IllegalAccessException e) {
			throw new DynamoDBMappingException(
					"Couldn't instantiate marshaller of class "
							+ marshallerClass, e);
		}

		return (T) marshaller.unmarshall(getter.getReturnType(), value.getS());
	}

	/**
	 * Returns an attribute value for the getter method with a custom marshaller
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	private AttributeValue getCustomerMarshallerAttributeValue(Method getter,
			Object getterReturnResult) {
		DynamoDBMarshalling annotation = getter
				.getAnnotation(DynamoDBMarshalling.class);
		Class<? extends DynamoDBMarshaller<? extends Object>> marshallerClass = annotation
				.marshallerClass();

		DynamoDBMarshaller marshaller;
		try {
			marshaller = marshallerClass.newInstance();
		} catch (InstantiationException e) {
			throw new DynamoDBMappingException(
					"Failed to instantiate custom marshaller for class "
							+ marshallerClass, e);
		} catch (IllegalAccessException e) {
			throw new DynamoDBMappingException(
					"Failed to instantiate custom marshaller for class "
							+ marshallerClass, e);
		}
		String stringValue = marshaller.marshall(getterReturnResult);

		return new AttributeValue().withS(stringValue);
	}

	/**
	 * Returns a marshaller that knows how to provide an AttributeValue for the
	 * result of the getter given.
	 */
	public ArgumentMarshaller getArgumentMarshaller(final Method getter) {

		synchronized (argumentMarshallerCache) {
			if (!argumentMarshallerCache.containsKey(getter)) {
				ArgumentMarshaller marshaller = null;

				if (isCustomMarshaller(getter)) {
					marshaller = new ArgumentMarshaller() {

						@Override
						public AttributeValue marshall(Object obj) {
							return getCustomerMarshallerAttributeValue(getter,
									obj);
						}
					};
				} else {

					Class<?> returnType = getter.getReturnType();
					if (Set.class.isAssignableFrom(returnType)) {
						Type genericType = getter.getGenericReturnType();
						if (genericType instanceof ParameterizedType) {
							returnType = (Class<?>) ((ParameterizedType) genericType)
									.getActualTypeArguments()[0];
						}

						if (Date.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									List<String> timestamps = new LinkedList<String>();
									for (Object o : (Set<?>) obj) {
										timestamps.add(new DateUtils()
												.formatIso8601Date((Date) o));
									}
									return new AttributeValue()
											.withSS(timestamps);
								}
							};
						} else if (Calendar.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									List<String> timestamps = new LinkedList<String>();
									for (Object o : (Set<?>) obj) {
										timestamps
												.add(new DateUtils()
														.formatIso8601Date(((Calendar) o)
																.getTime()));
									}
									return new AttributeValue()
											.withSS(timestamps);
								}
							};
						} else if (boolean.class.isAssignableFrom(returnType)
								|| Boolean.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									List<String> booleanAttributes = new ArrayList<String>();
									for (Object b : (Set<?>) obj) {
										if (b == null || !(Boolean) b) {
											booleanAttributes.add("0");
										} else {
											booleanAttributes.add("1");
										}
									}
									return new AttributeValue()
											.withNS(booleanAttributes);
								}
							};
						} else if (returnType.isPrimitive()
								|| Number.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									List<String> attributes = new ArrayList<String>();
									for (Object o : (Set<?>) obj) {
										attributes.add(String.valueOf(o));
									}
									return new AttributeValue()
											.withNS(attributes);
								}
							};
						} else {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									List<String> attributes = new ArrayList<String>();
									for (Object o : (Set<?>) obj) {
										attributes.add(String.valueOf(o));
									}
									return new AttributeValue()
											.withSS(attributes);
								}
							};
						}
					} else if (Collection.class.isAssignableFrom(returnType)) {
						throw new DynamoDBMappingException(
								"Non-set collections aren't supported: "
										+ (getter.getDeclaringClass() + "." + getter
												.getName()));
					} else {
						if (Date.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									return new AttributeValue()
											.withS(new DateUtils()
													.formatIso8601Date((Date) obj));
								}
							};
						} else if (Calendar.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									return new AttributeValue()
											.withS(new DateUtils()
													.formatIso8601Date(((Calendar) obj)
															.getTime()));
								}
							};
						} else if (boolean.class.isAssignableFrom(returnType)
								|| Boolean.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									if (obj == null || !(Boolean) obj) {
										return new AttributeValue().withN("0");
									} else {
										return new AttributeValue().withN("1");
									}
								}
							};
						} else if (returnType.isPrimitive()
								|| Number.class.isAssignableFrom(returnType)) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									return new AttributeValue().withN(String
											.valueOf(obj));
								}
							};
						} else if (returnType == String.class) {
							marshaller = new ArgumentMarshaller() {

								@Override
								public AttributeValue marshall(Object obj) {
									return new AttributeValue().withS(String
											.valueOf(obj));
								}
							};
						} else {
							throw new DynamoDBMappingException(
									"Unsupported type: " + returnType + " for "
											+ getter);
						}
					}
				}
				argumentMarshallerCache.put(getter, marshaller);
			}
		}

		return argumentMarshallerCache.get(getter);
	}

	/**
	 * Attempts to parse the string given as a boolean and return its value.
	 * Throws an exception if the value is anything other than 0 or 1.
	 */
	private boolean parseBoolean(String s) {
		if ("1".equals(s)) {
			return true;
		} else if ("0".equals(s)) {
			return false;
		} else {
			throw new IllegalArgumentException(
					"Expected 1 or 0 for boolean value, was " + s);
		}
	}

	/**
	 * Returns the attribute name corresponding to the given getter method.
	 */
	public String getAttributeName(Method getter) {
		synchronized (attributeNameCache) {
			if (!attributeNameCache.containsKey(getter)) {

				// First check for a hash key annotation
				DynamoDBHashKey hashKeyAnnotation = getter
						.getAnnotation(DynamoDBHashKey.class);
				if (hashKeyAnnotation != null
						&& hashKeyAnnotation.attributeName() != null
						&& hashKeyAnnotation.attributeName().length() > 0)
					return hashKeyAnnotation.attributeName();

				// Then a range key
				DynamoDBRangeKey rangeKey = getter
						.getAnnotation(DynamoDBRangeKey.class);
				if (rangeKey != null && rangeKey.attributeName() != null
						&& rangeKey.attributeName().length() > 0)
					return rangeKey.attributeName();

				// Then an attribute
				DynamoDBAttribute attribute = getter
						.getAnnotation(DynamoDBAttribute.class);
				if (attribute != null && attribute.attributeName() != null
						&& attribute.attributeName().length() > 0)
					return attribute.attributeName();

				// Finally a version attribute
				DynamoDBVersionAttribute version = getter
						.getAnnotation(DynamoDBVersionAttribute.class);
				if (version != null && version.attributeName() != null
						&& version.attributeName().length() > 0)
					return version.attributeName();

				// Default to method name
				String attributeName = null;
				if (getter.getName().startsWith("get")) {
					attributeName = getter.getName().substring("get".length());
				} else if (getter.getName().startsWith("is")) {
					attributeName = getter.getName().substring("is".length());
				} else {
					throw new DynamoDBMappingException(
							"Getter must begin with 'get' or 'is'");
				}

				// Lowercase the first letter of the name
				attributeName = attributeName.substring(0, 1).toLowerCase()
						+ attributeName.substring(1);
				attributeNameCache.put(getter, attributeName);
			}
		}

		return attributeNameCache.get(getter);
	}

	/**
	 * Returns the setter corresponding to the getter given, or null if no such
	 * setter exists.
	 */
	public Method getSetter(Method getter) {
		synchronized (setterCache) {
			if (!setterCache.containsKey(getter)) {
				String attributeName = null;
				if (getter.getName().startsWith("get")) {
					attributeName = getter.getName().substring("get".length());
				} else if (getter.getName().startsWith("is")) {
					attributeName = getter.getName().substring("is".length());
				} else {
					// should be impossible to reach this exception
					throw new RuntimeException(
							"Getter method must start with 'is' or 'get'");
				}
				String setterName = "set" + attributeName;
				Method setter = null;
				try {
					setter = getter.getDeclaringClass().getMethod(setterName,
							getter.getReturnType());
				} catch (NoSuchMethodException e) {
					throw new DynamoDBMappingException(
							"Expected a public, one-argument method called "
									+ setterName + " on class "
									+ getter.getDeclaringClass(), e);
				} catch (SecurityException e) {
					throw new DynamoDBMappingException(
							"No access to public, one-argument method called "
									+ setterName + " on class "
									+ getter.getDeclaringClass(), e);
				}
				setterCache.put(getter, setter);
			}
		}

		return setterCache.get(getter);
	}

	/**
	 * Returns a marshaller that knows how to provide an AttributeValue for the
	 * getter method given. Also increments the value of the getterReturnResult
	 * given.
	 */
	ArgumentMarshaller getVersionedArgumentMarshaller(final Method getter,
			Object getterReturnResult) {

		synchronized (versionArgumentMarshallerCache) {
			if (!versionArgumentMarshallerCache.containsKey(getter)) {

				ArgumentMarshaller marshaller = null;

				final Class<?> returnType = getter.getReturnType();
				if (BigInteger.class.isAssignableFrom(returnType)) {
					marshaller = new ArgumentMarshaller() {

						@Override
						public AttributeValue marshall(Object obj) {
							if (obj == null)
								obj = BigInteger.ZERO;
							Object newValue = ((BigInteger) obj)
									.add(BigInteger.ONE);
							return getArgumentMarshaller(getter).marshall(
									newValue);
						}
					};

				} else if (Integer.class.isAssignableFrom(returnType)) {
					marshaller = new ArgumentMarshaller() {

						@Override
						public AttributeValue marshall(Object obj) {
							if (obj == null)
								obj = new Integer(0);
							Object newValue = ((Integer) obj).intValue() + 1;
							return getArgumentMarshaller(getter).marshall(
									newValue);
						}
					};

				} else if (Byte.class.isAssignableFrom(returnType)) {
					marshaller = new ArgumentMarshaller() {

						@Override
						public AttributeValue marshall(Object obj) {
							if (obj == null)
								obj = new Byte((byte) 0);
							Object newValue = (byte) ((((Byte) obj).byteValue() + 1) % Byte.MAX_VALUE);
							return getArgumentMarshaller(getter).marshall(
									newValue);
						}
					};

				} else if (Long.class.isAssignableFrom(returnType)) {
					marshaller = new ArgumentMarshaller() {

						@Override
						public AttributeValue marshall(Object obj) {
							if (obj == null)
								obj = new Long(0);
							Object newValue = ((Long) obj).longValue() + 1L;
							return getArgumentMarshaller(getter).marshall(
									newValue);
						}
					};
				} else {
					throw new DynamoDBMappingException(
							"Unsupported parameter type for "
									+ DynamoDBVersionAttribute.class + ": "
									+ returnType
									+ ". Must be a whole-number type.");
				}

				versionArgumentMarshallerCache.put(getter, marshaller);
			}
		}

		return versionArgumentMarshallerCache.get(getter);
	}

	/**
	 * Returns a marshaller for the auto-generated key returned by the getter
	 * given.
	 */
	ArgumentMarshaller getAutoGeneratedKeyArgumentMarshaller(final Method getter) {
		synchronized (keyArgumentMarshallerCache) {
			if (!keyArgumentMarshallerCache.containsKey(getter)) {
				ArgumentMarshaller marshaller = null;

				Class<?> returnType = getter.getReturnType();
				if (String.class.isAssignableFrom(returnType)) {
					marshaller = new ArgumentMarshaller() {

						@Override
						public AttributeValue marshall(Object obj) {
							String newValue = UUID.randomUUID().toString();
							return getArgumentMarshaller(getter).marshall(
									newValue);
						}
					};
				} else {
					throw new DynamoDBMappingException(
							"Unsupported type for "
									+ getter
									+ ": "
									+ returnType
									+ ".  Only Strings are supported when auto-generating keys.");
				}

				keyArgumentMarshallerCache.put(getter, marshaller);
			}
		}

		return keyArgumentMarshallerCache.get(getter);
	}

	/**
	 * Returns whether the method given is an annotated, no-args getter of a
	 * version attribute.
	 */
	public boolean isVersionAttributeGetter(Method getter) {
		synchronized (versionAttributeGetterCache) {
			if (!versionAttributeGetterCache.containsKey(getter)) {
				versionAttributeGetterCache
						.put(getter,
								getter.getName().startsWith("get")
										&& getter.getParameterTypes().length == 0
										&& getter
												.getAnnotation(DynamoDBVersionAttribute.class) != null);
			}
		}

		return versionAttributeGetterCache.get(getter);
	}

	/**
	 * Returns whether the method given is an assignable key getter.
	 */
	public boolean isAssignableKey(Method getter) {
		synchronized (autoGeneratedKeyGetterCache) {
			if (!autoGeneratedKeyGetterCache.containsKey(getter)) {
				autoGeneratedKeyGetterCache
						.put(getter,
								getter.getAnnotation(DynamoDBAutoGeneratedKey.class) != null
										&& (getter
												.getAnnotation(DynamoDBHashKey.class) != null || getter
												.getAnnotation(DynamoDBRangeKey.class) != null));
			}
		}

		return autoGeneratedKeyGetterCache.get(getter);
	}

	/**
	 * Swallows the checked exceptions around Method.invoke and repackages them
	 * as {@link DynamoDBMappingException}
	 */
	public Object safeInvoke(Method method, Object object, Object... arguments) {
		try {
			return method.invoke(object, arguments);
		} catch (IllegalAccessException e) {
			throw new DynamoDBMappingException("Couldn't invoke " + method, e);
		} catch (IllegalArgumentException e) {
			throw new DynamoDBMappingException("Couldn't invoke " + method, e);
		} catch (InvocationTargetException e) {
			throw new DynamoDBMappingException("Couldn't invoke " + method, e);
		}

	}

	/**
	 * Gets the name of the attribute that represents the hashKey
	 */
	public <T> String getHashKeyAttributeName(Class<T> kindClass) {
		Method hashKeyGetter = getHashKeyGetter(kindClass);
		return getAttributeName(hashKeyGetter);
	}

	/**
	 * Gets the name of the attribute that represents the rangeKey
	 */
	public <T> String getRangeKeyAttributeName(Class<T> kindClass) {
		Method rangeKeyGetter = getRangeKeyGetter(kindClass);
		if (rangeKeyGetter != null) {
			return getAttributeName(rangeKeyGetter);
		}
		return null;
	}

	/**
	 * 
	 * Returns the attributeName for the given class fieldName. Returns null of
	 * not found in the class.
	 */
	public String fieldToAttributeName(Class<?> kindClass, String field) {
		synchronized (fieldToAttributeNameCache) {
			if (!fieldToAttributeNameCache.containsKey(kindClass)) {
				Collection<Method> relevantGetters = this
						.getRelevantGetters(kindClass);
				Map<String, String> fieldToAttributeNameMap = new HashMap<String, String>();
				fieldToAttributeNameCache.put(kindClass,
						fieldToAttributeNameMap);
				for (Method relevantGetter : relevantGetters) {
					String fieldName = null;
					if (relevantGetter.getName().startsWith("get")) {
						fieldName = relevantGetter.getName().substring(
								"get".length());
					} else if (relevantGetter.getName().startsWith("is")) {
						fieldName = relevantGetter.getName().substring(
								"is".length());
					} else {
						throw new DynamoDBMappingException(
								"Getter must begin with 'get' or 'is'");
					}
					// Lowercase the first letter of the name
					fieldName = fieldName.substring(0, 1).toLowerCase()
							+ fieldName.substring(1);
					fieldToAttributeNameMap.put(fieldName,
							getAttributeName(relevantGetter));
				}
			}
		}
		return fieldToAttributeNameCache.get(kindClass).get(field);
	}

	/**
	 * 
	 * Find the getter method for the given field
	 */
	public Method findGetterMethodForFieldName(Class<?> kindClass,
			String fieldName) {
		Collection<Method> relevantGetters = getRelevantGetters(kindClass);
		for (Method method : relevantGetters) {
			String attributeName = null;
			if (method.getName().startsWith("get")) {
				attributeName = method.getName().substring("get".length());
			} else if (method.getName().startsWith("is")) {
				attributeName = method.getName().substring("is".length());
			}
			if (attributeName != null) {
				attributeName = attributeName.substring(0, 1).toLowerCase()
						+ attributeName.substring(1);
			}
			if (attributeName.equals(fieldName)) {
				return method;
			}
		}
		throw new DynamoDBMappingException("Unknown field: " + fieldName);
	}

	/**
	 * 
	 * enum to represent S, N, SS, NS
	 * 
	 */
	public enum MarshallerType {
		S, N, SS, NS
	}

	/**
	 * 
	 * Determine whether the given getter is a (S, N, SS, NS). Returns null if
	 * method not found
	 */
	public MarshallerType getMarshallerType(Method getter) {
		synchronized (argumentMarshallerTypeCache) {
			if (!argumentMarshallerTypeCache.containsKey(getter)) {
				MarshallerType marshallerType = null;

				if (isCustomMarshaller(getter)) {

				} else {

					Class<?> returnType = getter.getReturnType();
					if (Set.class.isAssignableFrom(returnType)) {
						Type genericType = getter.getGenericReturnType();
						if (genericType instanceof ParameterizedType) {
							returnType = (Class<?>) ((ParameterizedType) genericType)
									.getActualTypeArguments()[0];
						}

						if (Date.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.SS;
						} else if (Calendar.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.SS;
						} else if (boolean.class.isAssignableFrom(returnType)
								|| Boolean.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.NS;
						} else if (returnType.isPrimitive()
								|| Number.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.NS;
						} else {
							marshallerType = MarshallerType.SS;
						}
					} else if (Collection.class.isAssignableFrom(returnType)) {
						throw new DynamoDBMappingException(
								"Non-set collections aren't supported: "
										+ (getter.getDeclaringClass() + "." + getter
												.getName()));
					} else {
						if (Date.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.S;
						} else if (Calendar.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.S;
						} else if (boolean.class.isAssignableFrom(returnType)
								|| Boolean.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.N;
						} else if (returnType.isPrimitive()
								|| Number.class.isAssignableFrom(returnType)) {
							marshallerType = MarshallerType.N;
						} else if (returnType == String.class) {
							marshallerType = MarshallerType.S;
						} else {
							throw new DynamoDBMappingException(
									"Unsupported type: " + returnType + " for "
											+ getter);
						}
					}
				}
				argumentMarshallerTypeCache.put(getter, marshallerType);
			}
		}

		return argumentMarshallerTypeCache.get(getter);
	}

	public AttributeValue getHashKeyElement(Object hashKey, Method hashKeyGetter) {
		AttributeValue hashKeyElement = new AttributeValue();
		Class<?> hashKeyMethodReturnType = hashKeyGetter.getReturnType();
		if (hashKeyMethodReturnType.isPrimitive()
				|| Number.class.isAssignableFrom(hashKeyMethodReturnType)) {
			hashKeyElement.setN(String.valueOf(hashKey));
		} else if (String.class.isAssignableFrom(hashKeyMethodReturnType)) {
			hashKeyElement.setS(String.valueOf(hashKey));
		} else {
			throw new DynamoDBMappingException(
					"Hash key property must be either a Number or a String");
		}
		return hashKeyElement;
	}

	public AttributeValue getRangeKeyElement(Object rangeKey,
			Method rangeKeyMethod) {
		AttributeValue rangeKeyElement = new AttributeValue();
		Class<?> rangeKeyMethodReturnType = rangeKeyMethod.getReturnType();
		if (rangeKeyMethodReturnType.isPrimitive()
				|| Number.class.isAssignableFrom(rangeKeyMethodReturnType)) {
			rangeKeyElement.setN(String.valueOf(rangeKey));
		} else if (String.class.isAssignableFrom(rangeKeyMethodReturnType)) {
			rangeKeyElement.setS(String.valueOf(rangeKey));
		} else {
			throw new DynamoDBMappingException(
					"Range key property must be either a Number or a String");
		}
		return rangeKeyElement;
	}

	/**
	 * Returns an {@link AttributeValue} corresponding to the getter and return
	 * result given, treating it as a non-versioned attribute. Only useful when
	 * differentiating between this method and getAttributeValue.
	 */
	public AttributeValue getSimpleAttributeValue(final Method getter,
			final Object getterReturnResult) {
		if (getterReturnResult == null)
			return null;

		ArgumentMarshaller marshaller = getArgumentMarshaller(getter);
		return marshaller.marshall(getterReturnResult);
	}

	/**
	 * Gets the attribute value object corresponding to the
	 * {@link DynamoDBVersionAttribute} getter, and its result, given. Null
	 * values are assumed to be new objects and given the smallest possible
	 * positive value. Non-null values are incremented from their current value.
	 */
	public AttributeValue getVersionAttributeValue(final Method getter,
			Object getterReturnResult) {
		ArgumentMarshaller marshaller = getVersionedArgumentMarshaller(getter,
				getterReturnResult);
		return marshaller.marshall(getterReturnResult);
	}

	public Collection<Method> getOnReadMethods(Class<?> clazz) {
		setupLifecycleCallbacks(clazz);
		return onReadMethodCache.get(clazz);
	}
	
	public Collection<Method> getOnCreateMethods(Class<?> clazz) {
		setupLifecycleCallbacks(clazz);
		return onCreateMethodCache.get(clazz);
	}
	
	public Collection<Method> getOnUpdateMethods(Class<?> clazz) {
		setupLifecycleCallbacks(clazz);
		return onUpdateMethodCache.get(clazz);
	}
	
	public Collection<Method> getOnDeleteMethods(Class<?> clazz) {
		setupLifecycleCallbacks(clazz);
		return onDeleteMethodCache.get(clazz);
	}
	
	private void setupLifecycleCallbacks(Class<?> clazz) {
		synchronized (clazz) {
			if (!onReadMethodCache.containsKey(clazz)) {
				onReadMethodCache.put(clazz, new ArrayList<Method>());
				onCreateMethodCache.put(clazz, new ArrayList<Method>());
				onUpdateMethodCache.put(clazz, new ArrayList<Method>());
				onDeleteMethodCache.put(clazz, new ArrayList<Method>());
				processLifecycleCallbacks(clazz);
			}
		}
	}

	/**
	 * Recursive function which walks up the superclass hierarchy looking for
	 * lifecycle-related methods (@OnSave and @OnLoad).
	 */
	private void processLifecycleCallbacks(Class<?> clazz) {
		if ((clazz == null) || (clazz == Object.class))
			return;

		// Start at the top of the chain
		this.processLifecycleCallbacks(clazz.getSuperclass());

		// Check all the methods
		for (Method method : clazz.getDeclaredMethods()) {
			if (method.isAnnotationPresent(OnCreate.class)
					|| method.isAnnotationPresent(OnRead.class)
					|| method.isAnnotationPresent(OnUpdate.class)
					|| method.isAnnotationPresent(OnDelete.class)) {
				method.setAccessible(true);

				if (method.getParameterTypes().length > 0)
					throw new IllegalStateException(
							"@OnCreate, @OnRead, @OnUpdate and @OnDelete methods can have no parameters");

				if (method.isAnnotationPresent(OnCreate.class)) {
					onCreateMethodCache.get(clazz).add(method);
				}

				if (method.isAnnotationPresent(OnRead.class)) {
					onReadMethodCache.get(clazz).add(method);
				}
				
				if (method.isAnnotationPresent(OnUpdate.class)) {
					onUpdateMethodCache.get(clazz).add(method);
				}
				
				if (method.isAnnotationPresent(OnDelete.class)) {
					onDeleteMethodCache.get(clazz).add(method);
				}
			}

		}
	}
	

	public <T> DynamoDBTableInitialCapacities getTableInitialCapacities(Class<T> clazz) {
		return clazz.getAnnotation(DynamoDBTableInitialCapacities.class);
	}

}
